[CORE-16608] cluster_link: validate source SR reachability and target emptiness on link creation#31011
Conversation
ff5d59a to
c6cad96
Compare
There was a problem hiding this comment.
Pull request overview
Adds Schema Registry (SR) API-sync preflight validation to cluster-link creation so that both validate_only and real creates are gated on (1) source SR reachability/auth and (2) destination-context emptiness for the contexts that would be imported into. This extends existing link preflight checks (Kafka broker probing) and introduces new SR client construction utilities plus unit/integration coverage.
Changes:
- Introduces SR preflight checker (
sr_preflight_checker) and wires it intocluster_link::manager::link_preflight_checks, surfacing failures asFAILED_PRECONDITION. - Adds a shared SR REST client builder (
schema_registry_sync::make_source_sr_client) + context-filter helper (filter_selects_source_context) to keep preflight logic consistent with runtime SR behavior. - Adds new C++ unit tests + ducktape end-to-end tests for SR preflight scenarios (unreachable source, non-empty target, mapping/filter scoping).
Reviewed changes
Copilot reviewed 37 out of 37 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tools/type-checking/type-check-strictness.json | Adds new ducktape test to strict type-check set. |
| tests/rptest/tests/shadow_link_sr_preflight_test.py | E2E ducktape coverage for SR reachability + target emptiness checks on validate/create. |
| src/v/schema/tests/fake_registry.h | Extends fake registry interface with has_subjects. |
| src/v/schema/tests/fake_registry.cc | Implements has_subjects for fake registry store. |
| src/v/schema/registry.h | Adds registry::has_subjects API for efficient emptiness checks. |
| src/v/schema/registry.cc | Implements has_subjects in enabled/disabled schema registry wrappers. |
| src/v/redpanda/admin/services/shadow_link/shadow_link.cc | Updates validate-only path to pass full metadata into test_connection. |
| src/v/redpanda/admin/services/shadow_link/err.cc | Maps new SR preflight errors to FAILED_PRECONDITION. |
| src/v/pandaproxy/schema_registry/rest_client/BUILD | Expands visibility so cluster_link can depend on SR REST client code. |
| src/v/datalake/tests/record_schema_resolver_test.cc | Updates test registry wrapper to implement has_subjects. |
| src/v/cluster/cluster_link/frontend.cc | Reuses shared context-filter predicate from cluster_link::model. |
| src/v/cluster/BUILD | Adds Bazel dep on new sr_context_mapping lib. |
| src/v/cluster_link/utils.h | Exposes TLS config builder used by both Kafka and SR client paths. |
| src/v/cluster_link/utils.cc | Refactors TLS config creation to support SR client construction. |
| src/v/cluster_link/tests/link_test.cc | Adds unit tests for SR preflight behavior and required Kafka API advertisement in mocks. |
| src/v/cluster_link/tests/deps.h | Adds fake_source_sr_prober and test fixture SR registry state. |
| src/v/cluster_link/tests/deps.cc | Wires SR preflight checker into manager fixture construction. |
| src/v/cluster_link/tests/BUILD | Adds deps for SR types + fake registry and additional Kafka protocol APIs. |
| src/v/cluster_link/sr_preflight_checker.h | Defines SR preflight checker and source SR prober interfaces. |
| src/v/cluster_link/sr_preflight_checker.cc | Implements SR reachability probe + destination-context emptiness validation. |
| src/v/cluster_link/service.h | Changes test_connection to accept full model::metadata for richer preflight. |
| src/v/cluster_link/service.cc | Constructs destination schema registry earlier and injects SR preflight checker into manager. |
| src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h | Updates delegating registry fixture to implement has_subjects. |
| src/v/cluster_link/schema_registry_sync/tests/source_client_test.cc | Unit tests for URL/TLS/auth parameter derivation for SR source client. |
| src/v/cluster_link/schema_registry_sync/tests/BUILD | Adds Bazel target for new source_client_test. |
| src/v/cluster_link/schema_registry_sync/source_client.h | Adds SR source client parameter resolution + client factory API. |
| src/v/cluster_link/schema_registry_sync/source_client.cc | Implements SR source URL parsing, TLS/SNI resolution, and client construction. |
| src/v/cluster_link/schema_registry_sync/BUILD | Adds Bazel library target for source_client. |
| src/v/cluster_link/model/sr_context_mapping.h | Declares shared predicate for SR context filter membership. |
| src/v/cluster_link/model/sr_context_mapping.cc | Implements shared predicate used by runtime and preflight. |
| src/v/cluster_link/model/BUILD | Adds Bazel library target for sr_context_mapping. |
| src/v/cluster_link/manager.h | Injects SR preflight checker dependency; updates test_connection signature. |
| src/v/cluster_link/manager.cc | Runs SR preflight after Kafka preflight and updates validate-only logic to use metadata. |
| src/v/cluster_link/fwd.h | Adds forward declaration for sr_preflight_checker. |
| src/v/cluster_link/errc.h | Adds new SR-related error codes. |
| src/v/cluster_link/errc.cc | Adds messages for new SR-related error codes. |
| src/v/cluster_link/BUILD | Adds new sources and deps for SR preflight and SR client construction. |
c6cad96 to
931d73a
Compare
|
force-pushed to resolve a merge conflict |
bartoszpiekny-redpanda
left a comment
There was a problem hiding this comment.
Code looks great but I'm not fully familiar with the functionality so please wait for someone else accepting it 🙂
| @@ -0,0 +1,72 @@ | |||
| /* | |||
There was a problem hiding this comment.
I've skipped reviewing this commit, because I think we'll be able to just drop in the source reader client from #30999 instead of this one.
| @@ -0,0 +1,30 @@ | |||
| /* | |||
There was a problem hiding this comment.
Nice! Thanks for splitting this into a separate commit 🚀
| # The API-sync path is feature-gated; ensure it is active so the | ||
| # request reaches the preflight checks rather than the feature gate. | ||
| self.target_cluster_service.set_feature_active(SR_API_SYNC_FEATURE, True) |
There was a problem hiding this comment.
I don't think this bit is necessary here.
The way these features work is that they auto-activate when all of the brokers are at least as up to date as a specific broker version. For this api sync feature specifically, the feature auto-activates when the brokers are all on at least version 26.2.
Since this tests creates a cluster where all brokers are at least 26.2, there is no need to manually enable the feature, it will just be auto-enabled while the cluster starts up.
| // Identity/unset mapping: the in-scope source contexts (those the link's | ||
| // filter selects) become the destination contexts. Filter membership is | ||
| // resolved through the shared predicate so this stays consistent with the | ||
| // runtime client-write blocker (cluster/cluster_link/frontend.cc). | ||
| auto identity_targets = [&] { | ||
| return source_contexts | std::views::filter([&](const auto& ctx) { | ||
| return model::filter_selects_source_context(api.filter, ctx); | ||
| }) | ||
| | std::views::transform([](const auto& ctx) { return ctx(); }) | ||
| | std::ranges::to<absl::flat_hash_set<ss::sstring>>(); | ||
| }; |
There was a problem hiding this comment.
I think for identity mapping we need to be slightly stricter than this. We should also validate that there is no context .ctx which doesn't exist in the source + exists in the destination + in scope in the source.
Unless I'm missing something, I think we won't need to consider the contents of the source SR at all, and these emptiness checks will just depend on the state of the config + destination schema registry.
There was a problem hiding this comment.
Done. Identity emptiness now reads the destination (list_subject_versions filtered by source scope) and never consults the source. An in-scope context present in the destination but absent from the source is now caught.
| } catch (...) { | ||
| eptr = std::current_exception(); | ||
| } | ||
| co_await client->shutdown(); |
There was a problem hiding this comment.
Can this shutdown throw? Wondering if we need to move it into the try-catch above
There was a problem hiding this comment.
Good catch — it can. Now wrapped in ss::coroutine::as_future, a failed close is logged and ignored so it can't mask the probe outcome.
931d73a to
ce6f252
Compare
|
force-pushed: review from @pgellert and local Claude incorporated
|
nguyen-andrew
left a comment
There was a problem hiding this comment.
lgtm, will review again once rebased and merged with Gellert's latest stuff
| // the destination context and the filter. | ||
| ss::future<chunked_vector<ss::sstring>> | ||
| identity_offending_contexts(const shadow_schema_registry_api& api) { | ||
| auto matches = co_await _destination.list_subject_versions( |
There was a problem hiding this comment.
nit: do we need all the subject versions here, or would just the subjects suffice? I think in the common case the destination should be empty or have very few subjects/subject-versions, so it probably wouldn't make a difference.
There was a problem hiding this comment.
Added get_subjects to the schema::registry interface (delegates to the existing sharded_store::get_subjects) and the identity check now lists subjects instead of subject-versions.
Done in af9f868
| # A source subject puts the default context in scope for identity | ||
| # mapping, so the check actually queries the (empty) target context. |
There was a problem hiding this comment.
IIUC, your latest push made the emptiness check derive its scope from the link config and only reads the destination registry (and not the source), right? If so, is this comment still accurate?
There was a problem hiding this comment.
Right, it's stale - removed it. With the config-scoped, destination-only check the source seed had no effect, so I dropped both the comment and the seeding.
| # A source subject puts the default context in scope for identity | ||
| # mapping; the target is left empty so the emptiness check passes. |
There was a problem hiding this comment.
Comment and seed removed.
ce6f252 to
7279c68
Compare
|
force-pushed: applied @nguyen-andrew review comments
Didn't rebase on @pgellert changes yet, will do next. |
Adds a short-circuiting emptiness probe forwarding to the existing sharded_store::has_subjects, plus implementations in every registry subclass so the interface stays pure-virtual.
Adds a subject listing forwarding to the existing sharded_store::get_subjects, returning one entry per subject rather than per (subject, version), plus implementations in every registry subclass so the interface stays pure-virtual.
Lifts filter_selects_source_context out of frontend.cc into a shared model target so the runtime client-write blocker and the preflight check resolve filter membership the same way. Behavior-preserving.
7279c68 to
51f1e07
Compare
|
force-pushed:
All changes incorporated in be8b85c (didn't include the exact diff because of the force-push, sorry for noisy diff) |
Retry command for Build#86796please wait until all jobs are finished before running the slash command |
|
/ci-repeat 1 |
Retry command for Build#86798please wait until all jobs are finished before running the slash command |
Extends manager::link_preflight_checks (already probing the remote Kafka brokers) with two Schema Registry API-sync checks that gate both the validate_only dry run and a real create: source reachability (link_sr_unreachable) and target context emptiness (link_sr_target_not_empty, scoped to the mapped destination contexts). Both surface to the admin ConnectRPC as FAILED_PRECONDITION. Adapt the existing SR-sync recovery E2E test to this behavior: creating a link against an unreachable source is now rejected up front, so the test can no longer point a fresh link at a bad URL. It instead creates the link against the live source, lets the first full sync land, then severs the destination's egress to the source SR mid-sync (firewall_blocked) to exercise the unavailable-then-recover path.
Two-cluster coverage of the SR preflight over validate_only and real create, positive and negative, asserting the specific failure message.
51f1e07 to
4dcf8e9
Compare
|
force-pushed: fix SR-sync recovery E2E test
|
Extends the shadow-link preflight (
manager::link_preflight_checks, whichalready probes the remote Kafka brokers) with two Schema Registry API-sync
checks. They run on the same path, so they gate both the
validate_onlydryrun and a real create:
settings, and credentials and list against it (
link_sr_unreachable).must be empty, scoped to the mapped contexts (
link_sr_target_not_empty).Both surface to the admin ConnectRPC as
FAILED_PRECONDITION.Backports Required
Release Notes